package com.yifeng.repo.flink.data.transport.sink;

import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.SQLException;
import java.util.ArrayList;
import java.util.List;

import org.apache.commons.collections.CollectionUtils;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.alibaba.fastjson.JSONObject;
import com.mysql.cj.util.StringUtils;
import com.yifeng.repo.flink.data.transport.config.DataSourceConfig;
import com.yifeng.repo.flink.data.transport.config.DataSourceConfigFactory;
import com.yifeng.repo.flink.data.transport.config.DdlConstants;
import com.yifeng.repo.flink.data.transport.config.SinkDbConfig;
import com.yifeng.repo.flink.data.transport.dto.AlterContent;
import com.yifeng.repo.flink.data.transport.dto.DataRow;
import com.yifeng.repo.flink.data.transport.dto.DbTypeEnums;
import com.yifeng.repo.flink.data.transport.dto.DdlContent;
import com.yifeng.repo.flink.data.transport.dto.DdlDataPacket;
import com.yifeng.repo.flink.data.transport.dto.DdlTypeEnums;
import com.yifeng.repo.flink.data.transport.relational.connection.JdbcConnectionFactory;

/**
 * ClassName: OracleDdlWriteToOracleSink
 * Description: 写入目标数据库sink
 * 该类仅仅用于测试DDL数据同步到Oracle
 */
public class DdlWriteToOracleSink extends RichSinkFunction<String> {
	private static final long serialVersionUID = 1L;

	private static final Logger LOG = LoggerFactory.getLogger(DdlWriteToOracleSink.class);
	
	private SinkDbConfig sinkDbConfig;
	
	public DdlWriteToOracleSink(SinkDbConfig sinkDbConfig) {
		this.sinkDbConfig = sinkDbConfig;
	}
	
	@Override
	public void invoke(String data, Context context) throws Exception {
		DataRow dataRow = JSONObject.parseObject(data, DataRow.class);
		//DDL类型下面的逻辑
		if (DdlConstants.SCHEMA_CHANGE_OP_TYPE.equals(dataRow.getOp().name())) {
			dealDDL(data);
			return;
		}
		dealDml(data);
	}
	
	/**
	 * 处理DML的逻辑
	 * @param data 监听到变更的数据
	 */
	private void dealDml(String data) {
	}

	/**
	 * 处理DDL的逻辑
	 * @param data 监听到变更的数据
	 * @throws SQLException SQL异常
	 */
	private void dealDDL(String data) throws SQLException {
		if(StringUtils.isNullOrEmpty(sinkDbConfig.getHostname())) {
			return;
		}
		DdlDataPacket ddlDataPacket = JSONObject.parseObject(data, DdlDataPacket.class);
		List<DdlContent> ddlContentList = ddlDataPacket.getDdlContentList();
		if(CollectionUtils.isEmpty(ddlContentList)) {
			return;
		}
		String tableName = ddlDataPacket.getTb();
		//根据设计判断是否需要转换表名
		List<String> ddlList = new ArrayList<String>();
		ddlContentList.forEach(ddlContent -> {
			//将ddlType转换成目标数据库的类型
			String ddlType = ddlContent.getDdlType();
			List<AlterContent> contentList = ddlContent.getContentList();
			contentList.forEach(alterContent -> {
				//拼接DDL
				String type = alterContent.getType();
				//根据数据库类型判断是否需要转换字段类型type
				StringBuffer ddl = new StringBuffer();
				if(DdlTypeEnums.ADD.getType().equals(ddlType) || DdlTypeEnums.DROP.getType().equals(ddlType) || DdlTypeEnums.MODIFY.getType().equals(ddlType)) {
					ddl.append("alter table " + tableName + " ");
				}
				if(DdlTypeEnums.ADD.getType().equals(ddlType)) {
					//根据字段类型判断是否需要添加小数位
					ddl.append("add (" + alterContent.getName() + " " + type + "(" + alterContent.getLength() + "))");
				}
				if(DdlTypeEnums.DROP.getType().equals(ddlType)) {
					ddl.append("drop (" + alterContent.getName() + ")");
				}
				if(DdlTypeEnums.MODIFY.getType().equals(ddlType)) {
					//根据字段类型判断是否需要添加小数位
					ddl.append("modify (" + alterContent.getName() + " " + type + "(" + alterContent.getLength() + "))");
				}
				//更换字段名
				//更换表名
				if(!StringUtils.isNullOrEmpty(ddl.toString())) {
					ddlList.add(ddl.toString());
				}
			});
		});
		//执行DDL语句
		if(CollectionUtils.isEmpty(ddlList)) {
			return;
		}
		DataSourceConfig dataSourceConfig = new DataSourceConfigFactory().create();
		dataSourceConfig.setDbType(DbTypeEnums.ORACLE.getType());
		//根据dbType确认driverClassName
    	dataSourceConfig.setDriverClassName("oracle.jdbc.OracleDriver");
    	//根据库名确认对应的hostname和port、database、用户名和密码
    	dataSourceConfig.setHostname(sinkDbConfig.getHostname());
    	dataSourceConfig.setPort(sinkDbConfig.getPort());
    	dataSourceConfig.setDatabase(sinkDbConfig.getDatabase());
    	dataSourceConfig.setUsername(sinkDbConfig.getUsername());
    	dataSourceConfig.setPassword(sinkDbConfig.getPassword());
		JdbcConnectionFactory factory = new JdbcConnectionFactory(dataSourceConfig);
		Connection connection = factory.connect(null);
		ddlList.forEach(ddl -> {
			try {
				PreparedStatement ps = connection.prepareStatement(ddl);
				int count = ps.executeUpdate();
				System.out.println("DDL:" + ddl + "是否执行成功: "+ count);
			} catch (SQLException e) {
				System.out.println("执行DDL语句异常：" + e.getMessage());
			}
		});
		connection.close();
	}
	
	@Override
    public void close() throws Exception {
    }
	
}
